package com.lizhiyu.flink.demo5_window;

import com.alibaba.fastjson.JSON;
import com.lizhiyu.flink.model.VideoOrder;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Objects;

/**
 * 增量统计  aggregate  来一条数据则统计一条数据 （不像之前的sum是等到了时间才会去统一统计）
 * aggregate 可以自定义聚合，比 sum 是固定统计，这个可以自定义统计规则
 */
public class WindowDemo2Aggregate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> ds = env.addSource(new CustomSource3());
        ds.setParallelism(1);
        ds.print("------------source:");
        //对数据格式进行转换
        SingleOutputStreamOperator<Tuple3<String, Integer,String>> mapDs = ds.map(new MapFunction<String, Tuple3<String, Integer,String>>() {
            @Override
            public Tuple3<String, Integer,String> map(String value) throws Exception {
                String[] split = value.split(":");
                Tuple3 Tuple3 = new Tuple3(split[0],Integer.parseInt(split[1]),split[2]);
                return Tuple3;
            }
        });

        //对数据进行分组聚合
        KeyedStream<Tuple3<String, Integer,String>, String> keyByDs = mapDs.keyBy(new KeySelector<Tuple3<String, Integer,String>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer,String> value) throws Exception {
                return value.f0;
            }
        });


        //额哥瑞 gay 特
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> aggregate = keyByDs.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                //aggregate  增量聚合函数  (对窗口内收集的数据做计算操作)
                .aggregate(
                //<IN, ACC, OUT>   in 输入值    acc 累加的值   out 返回的值
                new AggregateFunction<Tuple3<String, Integer, String>, VideoOrder, Tuple3<String, Integer, String>>() {

                    //初始化累加器
                    @Override
                    public VideoOrder createAccumulator() {
                        return new VideoOrder();
                    }

                    //进行累加操作
                    @Override
                    public VideoOrder add(Tuple3<String, Integer, String> value, VideoOrder accumulator) {
                        //使用aggregate 每当来一条数据则执行一次这个方法
                        System.out.println(JSON.toJSONString(value) + "执行了一次聚合");
                        accumulator.setMoney(value.f1 + accumulator.getMoney());
                        accumulator.setTitle(value.f0);
                        if (Objects.isNull(accumulator.getCreateTimeStr())) {
                            accumulator.setCreateTimeStr(value.f2);
                        }
                        return accumulator;
                    }

                    //获取结果
                    @Override
                    public Tuple3<String, Integer, String> getResult(VideoOrder accumulator) {
                        Tuple3<String, Integer, String> tuple3 = new Tuple3<>();
                        tuple3.f0 = accumulator.getTitle();
                        tuple3.f1 = accumulator.getMoney();
                        tuple3.f2 = accumulator.getCreateTimeStr();
                        return tuple3;
                    }

                    @Override
                    public VideoOrder merge(VideoOrder a, VideoOrder b) {
                        //合并内容一般不用
                        return null;
                    }
                }
        );

        aggregate.print("sink:");
        env.execute();
    }
}
